/**
 * Copyright [2020] [LiBo/Alex of copyright liboware@gmail.com ]
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.hyts.standard.webflux.react.event;

import cn.hutool.core.lang.Console;
import cn.hutool.core.lang.UUID;
import cn.hutool.core.thread.ThreadUtil;
import com.google.common.base.Preconditions;
import com.google.common.collect.Maps;
import org.reactivestreams.Subscription;
import reactor.core.CoreSubscriber;
import reactor.core.Disposable;
import reactor.core.publisher.TopicProcessor;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;
import reactor.util.concurrent.WaitStrategy;

import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.function.Consumer;
import java.util.function.Supplier;

/**
 * @project-name:standard-webflux
 * @package-name:com.hyts.standard.webflux.react.event
 * @author:LiBo/Alex
 * @create-date:2022-08-01 23:40
 * @copyright:libo-alex4java
 * @email:liboware@gmail.com
 * @description:
 */
public class EventBusFactory {


    /**
     *
     */
    private static final Map<String,EventBus> topicProcessorMap = Maps.newConcurrentMap();


    /**
     * 创建topicProcessor
     * @param name
     * @param bufferSize
     * @param shared
     * @param autoCanceled
     * @param executorService
     * @param requestExecuteService
     * @param waitStrategy
     * @param supplier
     * @param <T>
     * @return
     */
    static <T> TopicProcessor createProducer(String name,
                               int bufferSize,
                               boolean shared,
                               boolean autoCanceled,
                               ExecutorService executorService,
                               ExecutorService requestExecuteService,
                               WaitStrategy waitStrategy,
                               Supplier<T> supplier){
        TopicProcessor.Builder topicProcessorBuilder = TopicProcessor.builder();
        return topicProcessorBuilder.name(name).bufferSize(bufferSize).share(shared).
                autoCancel(autoCanceled).executor(executorService).
                requestTaskExecutor(requestExecuteService).
                waitStrategy(waitStrategy).signalSupplier(supplier).build();
    }

    /**
     * 创建topicProcessor
     * @param name
     * @param bufferSize
     * @param autoCanceled
     * @param executorService
     * @param requestExecuteService
     * @param waitStrategy
     * @param supplier
     * @param <T>
     * @return
     */
    public static <T> EventBus createProducer(String name,
                                                    int bufferSize,
                                                    boolean autoCanceled,
                                                    ExecutorService executorService,
                                                    ExecutorService requestExecuteService,
                                                    WaitStrategy waitStrategy,
                                                    Supplier<T> supplier){
        return topicProcessorMap.putIfAbsent(name,new EventBus(createProducer(name,bufferSize,Boolean.FALSE,
                autoCanceled,executorService,requestExecuteService,waitStrategy,supplier)));
    }

    /**
     * 创建topicProcessor
     * @param name
     * @param bufferSize
     * @param executorService
     * @param requestExecuteService
     * @param waitStrategy
     * @param supplier
     * @param <T>
     * @return
     */
    public static <T> EventBus createProducer(String name,
                                                    int bufferSize,
                                                    ExecutorService executorService,
                                                    ExecutorService requestExecuteService,
                                                    WaitStrategy waitStrategy,
                                                    Supplier<T> supplier){
        topicProcessorMap.putIfAbsent(name,
                        new EventBus(createProducer(name,bufferSize,
                                Boolean.FALSE,Boolean.FALSE,executorService,
                                requestExecuteService,waitStrategy,supplier)));
        return topicProcessorMap.get(name);
    }


    /**
     * 创建topicProcessor
     * @param name
     * @param executorService
     * @param requestExecuteService
     * @param waitStrategy
     * @param supplier
     * @param <T>
     * @return
     */
    public static <T> EventBus createProducer(String name,
                                                    ExecutorService executorService,
                                                    ExecutorService requestExecuteService,
                                                    WaitStrategy waitStrategy,
                                                    Supplier<T> supplier){
        topicProcessorMap.putIfAbsent(name,
                new EventBus(createProducer(name,2<<10,Boolean.FALSE,Boolean.FALSE,
                        executorService,requestExecuteService,waitStrategy,supplier)));
        return topicProcessorMap.get(name);
    }


    /**
     * 创建topicProcessor
     * @param name
     * @param requestExecuteService
     * @param waitStrategy
     * @param supplier
     * @param <T>
     * @return
     */
    public static <T> EventBus createProducer(String name,
                                                    ExecutorService requestExecuteService,
                                                    WaitStrategy waitStrategy,
                                                    Supplier<T> supplier){
        topicProcessorMap.putIfAbsent(name,
                new EventBus(createProducer(name,2<<10,Boolean.FALSE,Boolean.FALSE,
                ThreadUtil.newExecutor(20,50,500),
                requestExecuteService,waitStrategy,supplier)));
        return topicProcessorMap.get(name);
    }

    /**
     * 创建topicProcessor
     * @param name
     * @param waitStrategy
     * @param supplier
     * @param <T>
     * @return
     */
    public static <T> EventBus createProducer(String name,
                                                    WaitStrategy waitStrategy,
                                                    Supplier<T> supplier){
        topicProcessorMap.putIfAbsent(name,
                new EventBus(createProducer(name,2<<10,Boolean.FALSE,Boolean.FALSE,
                ThreadUtil.newExecutor(20,50,500),
                ThreadUtil.newExecutor(20,50,500),
                waitStrategy,supplier)));
        return topicProcessorMap.get(name);
    }

    /**
     * 创建topicProcessor
     * @param name
     * @param supplier
     * @param <T>
     * @return
     */
    public static <T> EventBus createProducer(String name,
                                                    Supplier<T> supplier){
        topicProcessorMap.putIfAbsent(name,
                new EventBus(createProducer(name,2<<10,Boolean.FALSE,Boolean.FALSE,
                ThreadUtil.newExecutor(20,50,500),
                ThreadUtil.newExecutor(20,50,500),
                WaitStrategy.busySpin(),supplier)));
        return topicProcessorMap.get(name);
    }


    /**
     * 创建topicProcessor
     * @param name
     * @param <T>
     * @return
     */
    public static <T> EventBus createProducer(String name){
        topicProcessorMap.putIfAbsent(name,
                new EventBus(createProducer(name,2<<10,Boolean.FALSE,Boolean.TRUE,
                ThreadUtil.newExecutor(20,50,500),
                ThreadUtil.newExecutor(20,50,500),
                WaitStrategy.busySpin(),null)));
        return topicProcessorMap.get(name);
    }


    /**
     * 创建消费者
     * @param name
     * @param <T>
     * @return
     */
    public static <T> void createConsumer(String name, Consumer<T> consumer, Scheduler scheduler){
        EventBus eventBus = topicProcessorMap.get(name);
        Preconditions.checkNotNull(eventBus,"the topic name:"+name+"'s producer is not exist!");
        Disposable disposable = eventBus.getTopicProcessor().subscribeOn(scheduler).
                subscribe(consumer);
    }

    /**
     * 创建消费者
     * @param name
     * @param consumer
     * @param errorConsumer
     * @param <T>
     */
    public static <T> void createConsumer(String name, Consumer<T> consumer,Consumer<T> errorConsumer,Scheduler scheduler){
        EventBus eventBus = topicProcessorMap.get(name);
        Preconditions.checkNotNull(eventBus,"the topic name:"+name+"'s producer is not exist!");
        eventBus.getTopicProcessor().subscribeOn(scheduler).subscribe(consumer,errorConsumer);
    }

    /**
     * 创建消费者
     * @param name
     * @param consumer
     * @param errorConsumer
     * @param runnable
     * @param <T>
     */
    public static <T> void createConsumer(String name, Consumer<T> consumer,
                                          Consumer<T> errorConsumer,Runnable runnable,Scheduler scheduler){
        EventBus eventBus = topicProcessorMap.get(name);
        Preconditions.checkNotNull(eventBus,"the topic name:"+name+"'s producer is not exist!");
        eventBus.getTopicProcessor().subscribeOn(scheduler).subscribe(consumer,errorConsumer,runnable);
    }

    /**
     * 创建消费者
     * @param name
     * @param consumer
     * @param errorConsumer
     * @param runnable
     * @param subscriptionConsumer
     * @param scheduler
     * @param <T>
     */
    public static <T> void createConsumer(String name, Consumer<T> consumer,
                                          Consumer<T> errorConsumer,Runnable runnable,
                                          Consumer<Subscription> subscriptionConsumer,
                                          Scheduler scheduler){
        EventBus eventBus = topicProcessorMap.get(name);
        Preconditions.checkNotNull(eventBus,"the topic name:"+name+"'s producer is not exist!");
        eventBus.getTopicProcessor().
                subscribeOn(scheduler).subscribe(consumer,errorConsumer,runnable,subscriptionConsumer);
    }

    /**
     * 创建消费者
     * @param name
     * @param coreSubscriber
     * @param <T>
     */
    public static <T> void createConsumer(String name, CoreSubscriber<T> coreSubscriber,Scheduler scheduler){
        EventBus eventBus = topicProcessorMap.get(name);
        Preconditions.checkNotNull(eventBus,"the topic name:"+name+"'s producer is not exist!");
        eventBus.getTopicProcessor().subscribeOn(scheduler).subscribe(coreSubscriber);
    }



    public static void main(String[] args) {
        EventBus eventBus = EventBusFactory.createProducer("name");
//        EventBusFactory.createConsumer("name",param->{
//            System.out.println(param+"--"+Thread.currentThread().getName());
//        }, Schedulers.elastic());
        try {
            EventBusFactory.createConsumer("name",new EventSubscriber<>(),
                    Schedulers.newParallel("test"));
            eventBus.push("easda"+Thread.currentThread().getName());
            eventBus.push("easda"+Thread.currentThread().getName());
            eventBus.push("easda"+Thread.currentThread().getName());
            Thread.currentThread().join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }

}
